package com.ruicar.afs.cloud.batch.frame.control;


import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.ruicar.afs.cloud.batch.entity.AfsBatchInfo;
import com.ruicar.afs.cloud.batch.entity.AfsBatchJobGroupInfo;
import com.ruicar.afs.cloud.batch.entity.AfsBatchJobStepInfo;
import com.ruicar.afs.cloud.batch.enums.BatchType;
import com.ruicar.afs.cloud.batch.enums.StepType;
import com.ruicar.afs.cloud.batch.enums.SystemStatus;
import com.ruicar.afs.cloud.batch.frame.bean.BatchRunGroupInfo;
import com.ruicar.afs.cloud.batch.frame.bean.BatchRunStepInfo;
import com.ruicar.afs.cloud.batch.service.*;
import com.ruicar.afs.cloud.common.core.constant.CommonConstants;
import com.ruicar.afs.cloud.common.core.enums.AfsEnumUtil;
import com.ruicar.afs.cloud.common.core.exception.AfsBaseException;
import com.ruicar.afs.cloud.common.core.util.SpringContextHolder;
import com.ruicar.afs.cloud.common.data.datasource.annotation.DS;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert;

import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * The type BatchControl
 * <p>Description: </p>
 *
 * @author Fzero
 * @version 1.0
 * @date Created on 2020.08.07 10:55:09
 * @since 1.0
 */
@AllArgsConstructor
@Slf4j
@Component
@Data
public class BatchControl {
    private final AfsBatchInfoService afsBatchInfoService;
    private final AfsBatchLogGroupService afsBatchLogGroupService;
    private final AfsBatchJobGroupInfoService afsBatchJobGroupInfoService;
    private final AfsBatchJobStepInfoService afsBatchJobStepInfoService;
    private final AfsBatchLogStepService afsBatchLogStepService;
    public static void startBatch(String systemName,String batchNo){
        BatchControl batchControl =  SpringContextHolder.getBean(BatchControl.class);
        BatchRunGroupInfo batchRunGroupInfo = batchControl.loadAndChangeBatchStep(systemName,batchNo);
        MonitorControl monitorControl = new MonitorControl();
        monitorControl.setDaemon(true);
        monitorControl.setName("batch-monitor");
        monitorControl.start();
        BatchGroupControlRunner batchGroupControlRunner = new BatchGroupControlRunner(null,batchRunGroupInfo,batchControl.getAfsBatchLogGroupService(),batchControl.getAfsBatchLogStepService(),batchControl.getAfsBatchInfoService(),monitorControl);
        batchGroupControlRunner.setDaemon(true);
        batchGroupControlRunner.setName(MessageFormat.format("批量组-{0}-{1}-{2}",batchRunGroupInfo.getAfsBatchJobGroupInfo().getSystemName(),batchRunGroupInfo.getAfsBatchJobGroupInfo().getBatchNo(),batchRunGroupInfo.getAfsBatchJobGroupInfo().getBatchName()));
        batchGroupControlRunner.start();
    }

    public static void resetError(String systemName){
        SpringContextHolder.getBean(BatchControl.class).resetToNormal(systemName);
    }

    @Transactional(rollbackFor = Exception.class)
    @DS(CommonConstants._BASE_DS_NAME)
    public void resetToNormal(String systemName){
        //检测系统状态，处于批量中不允许再次发起批量
        AfsBatchInfo afsBatchInfo =  afsBatchInfoService.getSystemBatchInfo(systemName);
        if(afsBatchInfo==null){
            throw new AfsBaseException(MessageFormat.format("系统编号{0}不存在，请检查配置",systemName));
        }
        if(afsBatchInfo.getSystemStatus() == SystemStatus.BATCHING||afsBatchInfo.getLiveThreadNum()>0){
            throw new AfsBaseException("系统处于批量中不允许操作");
        }
        afsBatchInfoService.changeBatchToNormalConsole(systemName);
    }

    @Transactional(rollbackFor = Exception.class)
    @DS(CommonConstants._BASE_DS_NAME)
    public BatchRunGroupInfo loadAndChangeBatchStep(String systemName,String batchNo){
        //检测系统状态，处于批量中不允许再次发起批量
        AfsBatchInfo afsBatchInfo =  afsBatchInfoService.getSystemBatchInfo(systemName);
        if(afsBatchInfo==null){
            throw new AfsBaseException(MessageFormat.format("系统编号{0}不存在，请检查配置",systemName));
        }
        if(afsBatchInfo.getSystemStatus() == SystemStatus.BATCHING||afsBatchInfo.getLiveThreadNum()>0){
            throw new AfsBaseException("系统处于批量中不允许再次进入批量");
        }

        if(afsBatchInfo.getSystemStatus()==SystemStatus.ERROR){
            throw new AfsBaseException("系统处于异常状态请重置批量状态");
        }
        if(afsBatchJobGroupInfoService.checkHasDepends(systemName,batchNo)){
            throw new AfsBaseException(MessageFormat.format("批量号{0}有前置批量不允许从中间发起批量，请检查配置",batchNo));
        }
        afsBatchInfo.setSystemStatus(SystemStatus.BATCHING);
        afsBatchInfoService.updateById(afsBatchInfo);
        return loadGroupInfoFromDb(systemName,batchNo);
    }

    @DS(CommonConstants._BASE_DS_NAME)
    public BatchRunGroupInfo loadGroupInfoFromDb(String systemName,String batchNo){
        AfsBatchInfo afsBatchInfo =  afsBatchInfoService.getSystemBatchInfo(systemName);
        BatchRunGroupInfo batchRunGroupInfo = new BatchRunGroupInfo();
        List<AfsBatchJobGroupInfo> afsBatchJobGroupInfos = afsBatchJobGroupInfoService.list(Wrappers.<AfsBatchJobGroupInfo>lambdaQuery().eq(AfsBatchJobGroupInfo::getSystemName,systemName));
        List<AfsBatchJobStepInfo> afsBatchJobStepInfos = afsBatchJobStepInfoService.list(Wrappers.<AfsBatchJobStepInfo>lambdaQuery().eq(AfsBatchJobStepInfo::getSystemName,systemName).ne(AfsBatchJobStepInfo::getStepType, StepType.NOT_RUN));
        Assert.isTrue(afsBatchInfo.getBatchType() != null, "待运行批量类型为空请检查批量配置信息");
        log.info("===========待运行批量类型为[{}]===========", AfsEnumUtil.desc(afsBatchInfo.getBatchType()));
        //由于日切关系当状态为日初时只运行日初类型批量
        if (afsBatchInfo.getBatchType() == BatchType.DAY_BEGIN) {
            batchRunGroupInfo.setResume(true);
            afsBatchJobStepInfos = afsBatchJobStepInfos
                    .stream()
                    .filter(afsBatchJobStepInfo -> afsBatchJobStepInfo.getBatchType() == afsBatchInfo.getBatchType()).collect(Collectors.toList());
        }else{
            batchRunGroupInfo.setResume(false);
        }
        Map<String,List<AfsBatchJobStepInfo>> jobStepMap = afsBatchJobStepInfos.stream().collect(Collectors.groupingBy(AfsBatchJobStepInfo::getBatchNo));
        batchRunGroupInfo.setRootGroup(true);
        genBatchRunGroup(batchNo,batchRunGroupInfo, afsBatchJobGroupInfos,jobStepMap);
        return batchRunGroupInfo;
    }

    private void genBatchRunGroup(String batchNo, BatchRunGroupInfo batchRunGroupInfo , List<AfsBatchJobGroupInfo> afsBatchJobGroupInfos, Map<String,List<AfsBatchJobStepInfo>> jobStepMap){
        AfsBatchJobGroupInfo afsBatchJobGroupInfo =  afsBatchJobGroupInfos
                .stream()
                .filter(batchJobInfo -> batchJobInfo.getBatchNo().equalsIgnoreCase(batchNo)).findAny().orElseGet(null);
        if(afsBatchJobGroupInfo ==null){
            return;
        }
        batchRunGroupInfo.setAfsBatchJobGroupInfo(afsBatchJobGroupInfo);
        batchRunGroupInfo.setBatchRunStepInfos(new ArrayList<>());
        batchRunGroupInfo.setSubGroups(new ArrayList<>());
        jobStepMap.getOrDefault(batchNo,new ArrayList<>())
                .stream()
                .filter(step-> StringUtils.isEmpty(step.getDependStep()))
                .forEach(step->{
                    BatchRunStepInfo batchRunStepInfo = new BatchRunStepInfo();
                    batchRunStepInfo.setAfsBatchJobStepInfo(step);
                    batchRunStepInfo.setSubBatchSteps(new ArrayList<>());
                    fillSteps(step.getStepNo(),batchRunStepInfo,jobStepMap.getOrDefault(batchNo,new ArrayList<>()));
                    batchRunGroupInfo.getBatchRunStepInfos().add(batchRunStepInfo);
                });

        afsBatchJobGroupInfos
                .stream()
                .filter(batchJobInfo -> batchNo.equalsIgnoreCase(batchJobInfo.getDependBatchNo()))
                .forEach(job->{
                    BatchRunGroupInfo subBatchRunGroup = new BatchRunGroupInfo();
                    genBatchRunGroup(job.getBatchNo(),subBatchRunGroup, afsBatchJobGroupInfos,jobStepMap);
                    batchRunGroupInfo.getSubGroups().add(subBatchRunGroup);
                });
    }

    private void fillSteps(String stepNo,BatchRunStepInfo batchRunStepInfo , List<AfsBatchJobStepInfo> afsBatchJobStepInfoList){
        List<AfsBatchJobStepInfo> subSteps =  afsBatchJobStepInfoList.stream().filter(step->stepNo.equalsIgnoreCase(step.getDependStep())).collect(Collectors.toList());
        if(subSteps.size()==0){
            return;
        }
        subSteps.forEach(step->{
            BatchRunStepInfo subBatchRunStepInfo = new BatchRunStepInfo();
            subBatchRunStepInfo.setAfsBatchJobStepInfo(step);
            subBatchRunStepInfo.setSubBatchSteps(new ArrayList<>());
            fillSteps(step.getStepNo(), subBatchRunStepInfo, afsBatchJobStepInfoList);
            batchRunStepInfo.getSubBatchSteps().add(subBatchRunStepInfo);
        });
    }
}
